<?php

namespace App\Commons\Libs\KafkaPHP;

use Arr;

class Producer
{
    private $config;
    public function __construct($config)
    {
        $this->config = $config;
    }
    /**
     * 发送数据
     * @param  array  $data [{topic, value, key}]
     * @return [type]       [description]
     */
    public function send(string $topic, $value, string $key = '')
    {
        $data = [];
        if (is_array($topic)) {
            $data = $topic;
        } elseif (!empty($value)) {
            $value = json_encode($value, JSON_UNESCAPED_UNICODE);
            $data = [compact('topic', 'value', 'key')];
        } else {
            throw_on('生产者发送消息不能为空');
        }
        dump('producer', $data);
        $config = \Kafka\ProducerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs($this->config['metadata_refresh_interval_ms']);
        $config->setMetadataBrokerList($this->config['metadata_broker_list']);
        $config->setBrokerVersion($this->config['broker_version']);
        $config->setRequiredAck(1);
        $config->setIsAsyn($this->config['is_asyn']);
        $config->setProduceInterval($this->config['produce_interval']);
        $producer = new \Kafka\Producer;
        $producer->setLogger(logs('kafka-p'));
        // $producer->success(function ($result) {
        //     return "success";
        // });
        // $producer->error(function ($errorCode) {
        //     var_dump($errorCode);
        // });
        // $producer = new \Kafka\Producer;
        $result = $producer->send($data);

        $this->checkSuccess($result);
        return $result;
    }

    private function checkSuccess(array $result)
    {
        $result = $result[0];
        $partition = Arr::get($result, 'data.0.partitions.0');

        if (intval($partition['errorCode']) !== 0) {
            throw_e('生产者发送消息失败');
        }
    }
}
